"""
Case Type   : USTORE
Case Name   : 验证slotname不可以数字开头
Create At   : 2025/4/1
Owner       : wan005
Description :
    1.查看wal_level，enable_slot_log，enable_default_ustore_table的初始值
    2.修改wal_level=logical;enable_slot_log=on；enable_default_ustore_table=on
    3.配置白名单用户
    4.主机创建逻辑复制槽
    5.删除逻辑复制槽
Expect      :
    1.初始值分别对应为hot_standby和off，offshow wal_level;show enable_slot_log
    2.修改成功
    3.配置完成
    4.创建成功
    5.删除成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger
from testcase.utils.Constant import Constant


class Security(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f'---{os.path.basename(__file__)} start---')
        self.userNode = Node('PrimaryDbUser')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.slotname = '900s_replication'
        self.config = os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')
        self.config_bak = os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba_bak.conf')

    def test_ustore(self):
        text = '-----pre1:备份pg_hba.conf文件;expect:成功-----'
        self.logger.info(text)
        cp_cmd = f"cp {self.config} {self.config_bak}"
        self.logger.info(cp_cmd)
        cp_result = self.userNode.sh(cp_cmd).result()
        self.logger.info(cp_result)
        
        text = '-----step1:查询参数默认值;expect:成功-----'
        self.logger.info(text)
        res1_1 = self.common.show_param('wal_level')
        self.assertEqual('hot_standby', res1_1, text + '执行失败')
        res1_2 = self.common.show_param('enable_slot_log')
        self.assertEqual('off', res1_2, text + '执行失败')
        res1_3 = self.common.show_param('enable_default_ustore_table')
        self.assertEqual('off', res1_3, text + '执行失败')

        text = '----step2:修改参数值;expect:成功----'
        self.logger.info(text)
        self.sh_primy.execute_gsguc('set', self.constant.GSGUC_SUCCESS_MSG,
                                    'wal_level=logical')
        self.sh_primy.restart_db_cluster()
        res2_1 = self.common.show_param('wal_level')
        self.assertEqual('logical', res2_1, text + '执行失败')
        self.sh_primy.execute_gsguc('reload', self.constant.GSGUC_SUCCESS_MSG,
                                    'enable_slot_log=on')
        res2_2 = self.common.show_param('enable_slot_log')
        self.assertEqual('on', res2_2, text + '执行失败')
        self.sh_primy.execute_gsguc('reload', self.constant.GSGUC_SUCCESS_MSG,
                                    'enable_default_ustore_table=on')
        res2_3 = self.common.show_param('enable_default_ustore_table')
        self.assertEqual('on', res2_3, text + '执行失败')

        text = '----step3:配置白名单;expect:成功----'
        self.logger.info(text)
        param_1 = f'local replication {self.userNode.db_user}  trust'
        param_2 = f'host replication {self.userNode.db_user} 127.0.0.1/32 trust'
        param_3 = f'host replication {self.userNode.db_user} ::1/128 trust'
        self.sh_primy.execute_gsguc('reload', self.constant.GSGUC_SUCCESS_MSG,
                                    'enable_default_ustore_table=on')
        result3_1 = self.sh_primy.execute_gsguc('reload',
                    self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        self.assertTrue(result3_1, text + '执行失败')
        result3_2 = self.sh_primy.execute_gsguc('reload',
                    self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        self.assertTrue(result3_2, text + '执行失败')
        result3_3 = self.sh_primy.execute_gsguc('reload',
                    self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result3_3, text + '执行失败')

        text = '----step4:主机创建逻辑复制槽;expect:成功----'
        self.logger.info(text)
        result4 = self.sh_primy.exec_pg_recvlogical('--create',
                                    self.userNode.db_name, self.slotname,
                                    self.userNode.db_user, '-P mppdb_decoding',
                                    passwd=self.userNode.db_password)
        self.assertNotIn('FATAL', result4, text + '执行失败')
        self.assertNotIn('ERROR', result4, text + '执行失败')
        text = '----删除逻辑复制槽----'
        self.logger.info(text)
        del_res = self.sh_primy.exec_pg_recvlogical('--drop',
                                    self.userNode.db_name, self.slotname,
                                    self.userNode.db_user,
                                    passwd=self.userNode.db_password)
        self.assertNotIn('FATAL', del_res, text + '执行失败')
        self.assertNotIn('ERROR', del_res, text + '执行失败')

    def tearDown(self):
        text = '-----step1:清理环境；expect:成功-----'
        self.logger.info(text)
        self.logger.info('-----恢复配置文件-----')
        clean_cmd1 = f"cp -f {self.config_bak} {self.config};" \
                     f"rm -rf {self.config_bak};"
        self.logger.info(clean_cmd1)
        clean_res1 = self.userNode.sh(clean_cmd1).result()
        self.logger.info(clean_res1)
        self.logger.info('-----恢复配参数配置-----')
        self.sh_primy.execute_gsguc('set', self.constant.GSGUC_SUCCESS_MSG,
                                    'wal_level=hot_standby')
        self.sh_primy.restart_db_cluster()
        clean_res2 = self.common.show_param('wal_level')
        self.assertEqual('hot_standby', clean_res2, text + '执行失败')
        self.sh_primy.execute_gsguc('reload', self.constant.GSGUC_SUCCESS_MSG,
                                    'enable_slot_log=off')
        clean_res3 = self.common.show_param('enable_slot_log')
        self.assertEqual('off', clean_res3, text + '执行失败')
        self.sh_primy.execute_gsguc('reload', self.constant.GSGUC_SUCCESS_MSG,
                                    'enable_default_ustore_table=off')
        clean_res4 = self.common.show_param('enable_default_ustore_table')
        self.assertEqual('off', clean_res4, text + '执行失败')
        self.logger.info(f'---{os.path.basename(__file__)} end---')
